/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package geektimes.event.distributed;

import org.apache.activemq.ActiveMQConnectionFactory;
import geektimes.event.EventListener;
import geektimes.event.reactive.stream.ListenerSubscriberAdapter;
import geektimes.reactive.streams.SimplePublisher;

import javax.jms.*;
import java.util.EventObject;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Distributed {@link EventObject} Publisher
 *
 * @author <a href="mailto:mercyblitz@gmail.com">Mercy</a>
 * @since 1.0.0
 */
public class DistributedEventHandler {

    private final SimplePublisher<EventObject> simplePublisher;

    ConnectionFactory connectionFactory;
    //定义链接
    Connection connection;
    //定义会话
    Session session;
    //定义消息目的地
    Destination destination;
    //定义消息生产者
    MessageProducer mProducer;
    //定义消息消费者
    MessageConsumer mConsumer;
    //定义消息对象
    Message message;

    public DistributedEventHandler(String uri) throws JMSException {
        simplePublisher = new SimplePublisher();
        //创建连接对象
        connectionFactory = new ActiveMQConnectionFactory("admin", "admin", uri);
        connection = connectionFactory.createConnection();
        //启动连接
        connection.start();
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //创建目的地，也就是队列名
        destination = session.createQueue("mercyblitz_test");
        //创建消息生成者，该生成者与目的地绑定
        // Build-in listener
        addEventListener(event -> {
            if (event instanceof DistributedEventObject) {
                // Event -> Pub/Sub
                try {

                    mProducer = session.createProducer(destination);
                    //创建消息
                    message = session.createTextMessage("Hello, Stephanie");
                    //发送消息
                    mProducer.send(message);
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });
    }


    public void publish(Object event) {
        simplePublisher.publish(new DistributedEventObject(event));
    }

    private void publish(EventObject event) {
//        simplePublisher.publish(new EventObject(event));
        simplePublisher.publish(event);
    }

    public void addEventListener(EventListener eventListener) {
        simplePublisher.subscribe(new ListenerSubscriberAdapter(eventListener));
    }

    public void subscriber()
    {
        try {
            mConsumer = session.createConsumer(destination);
            message = mConsumer.receive();
            String msg = ((TextMessage)message).getText();
            publish(new EventObject(msg));
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    public void close() throws JMSException {
        mProducer.close();
        session.close();
        connection.close();
    }

}